package com.sdehl.kafkatransfer.kafka.consumer;

import com.alibaba.fastjson.JSONObject;
import com.sdehl.kafkatransfer.constant.KafkaConst;
import com.sdehl.kafkatransfer.entity.passcar.PassCarInfo;
import com.sdehl.kafkatransfer.kafka.producer.MyKafkaProducer;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Optional;

/**
 * @Author 18011618
 * @Description
 * @Date 14:50 2018/7/20
 * @Modify By
 */
@Component
@Slf4j
public class MyKafkaConsumer {
    @Autowired
    private MyKafkaProducer myKafkaProducer;

    /**
     * kafka消费消息
     * @param consumerRecord
     * 注意:这里groupId优先级比配置文件中高
     * https://blog.csdn.net/u010634066/article/details/109803987
     */
    @KafkaListener(topics = {KafkaConst.CONSUMER_TOPIC},groupId = "kafka-transfer-group")
    public void consumer(ConsumerRecord<?,?> consumerRecord){
        //判断是否为null
        Optional<?> kafkaMessage = Optional.ofNullable(consumerRecord.value());
        if(kafkaMessage.isPresent()){
            //得到Optional实例中的值
            Object message = kafkaMessage.get();
            log.info("kafka消费消息,topic:{},内容：{}",KafkaConst.CONSUMER_TOPIC , JSONObject.toJSONString(message));
            //fixme 解析文本，将内容解析成PassCarInfo对象
            PassCarInfo passCarInfo = transferMsgToPassCar(message);
            myKafkaProducer.send(passCarInfo);
        }
    }

    /**
     * 将消费数据转换为PassCarInfo
     * @param message
     * @return
     */
    private PassCarInfo transferMsgToPassCar(Object message){
        String realData = ((String)message).substring(5);
        String[] dataInfoArr = realData.split("\\\\n");
        JSONObject dataObj = new JSONObject();
        for(String dataInfo:dataInfoArr){
            String[] dataArr = dataInfo.split("=");
            String value = dataArr.length > 1 ? dataArr[1]:"";
            dataObj.put(dataArr[0],value);
        }
        return JSONObject.parseObject(dataObj.toJSONString(), PassCarInfo.class);
    }
}
